/*
Copyright 2011 Raymond Giorgi

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */

package edu.pitt.cs3530.frontend;

import java.io.IOException;

import javax.xml.parsers.ParserConfigurationException;

import org.xml.sax.SAXException;

import edu.pitt.cs3530.applicationcontrolling.ServerController;
import edu.pitt.cs3530.applicationcontrolling.ServerController.Scale;
import edu.pitt.cs3530.configuration.FrontEndConfiguration;
import edu.pitt.cs3530.controllers.PredictorListener;
import edu.pitt.cs3530.controllers.ReactorListener;
import edu.pitt.cs3530.database.DatabaseFiftyToHundred;
import edu.pitt.cs3530.predictors.MovingAverageThread;
import edu.pitt.cs3530.predictors.PredictorThread;
import edu.pitt.cs3530.reactors.ReactorThread;
import edu.pitt.cs3530.reactors.ReactorMadeMissed;
import edu.pitt.cs3530.shareddatastructures.Utilization;
import edu.pitt.cs3530.threads.ReceiverThread.ReceiveConnection;

/**
 * This class is run on the front end and implements both a predictor and a
 * reactor
 */
public class FrontProactiveMain extends FrontEnd implements ReactorListener,
		PredictorListener {

	private long proactiveElected;
	private long reactiveElected;
	private long timeToChangeScale;
	double changeSinceLastScale = 0.0;
	private int scaleDownCounter;

	public FrontProactiveMain() {
		// (new DeleteOldFilesThread("/tmp/", Constants.SLEEP_DELETE,
		// Constants.TIME_TO_DELETE)).start();
		super();
		proactiveElected = 0;
		reactiveElected = 0;
		this.timeToChangeScale = System.currentTimeMillis();
		String tempDirectory = FrontEndConfiguration.getInstance()
				.getTemporaryDirectory();
		long pastInterval = FrontEndConfiguration.getInstance()
				.getTimeToKeepRequests();
		int intervals = FrontEndConfiguration.getInstance()
				.getPredictorIntervals();
		long predictSleep = FrontEndConfiguration.getInstance()
				.getPredictorSleepTime();
		System.out.println(predictSleep);
		predictor = new MovingAverageThread(pastInterval, pastInterval,
				intervals, predictSleep, "SimpleQoS", null);
		predictor.addListener(this);
		predictor.start();

		long deadline = FrontEndConfiguration.getInstance().getLateDeadline();
		long earlyDeadline = FrontEndConfiguration.getInstance()
				.getEarlyDeadling();
		long responseSleep = FrontEndConfiguration.getInstance()
				.getReactorSleepTime();

		response = new ReactorMadeMissed(deadline, earlyDeadline, responseSleep,
				pastInterval / 2, "SimpleQoS", null);
		response.addListener(this);
		response.start();
		new DeleteOldFilesThread(tempDirectory, pastInterval, pastInterval)
				.start();
		new DatabaseFiftyToHundred("aleksi.cs.pitt.edu").start();
	}

	private static FrontProactiveMain _instance;

	public static FrontProactiveMain Instance() {
		if (_instance == null) {
			_instance = new FrontProactiveMain();
		}
		return _instance;
	}

	public static void main(String args[]) throws SAXException,
			ParserConfigurationException, IOException {
		buildConfiguration(args[0]);
		new ReceiveConnection().start();
		FrontProactiveMain.Instance();
	}

	@Override
	public synchronized void alertChange(PredictorThread predictor) {
		long currentTime = System.currentTimeMillis();
		if (timeToChangeScale < currentTime) {
			double upperBound = FrontEndConfiguration.getInstance()
					.getPredictorUpperUtilization();
			double lowerBound = FrontEndConfiguration.getInstance()
					.getPredictorLowerScaleUtilization();
			double extemeLowerBound = FrontEndConfiguration.getInstance()
					.getPredictorLowerOffUtilization();
			double targetQos = (upperBound + lowerBound) / 2.0;
			Scale scale = Scale.NONE;
			ServerController controller = ServerController.getInstance();
			double change = predictor.predictedChange(FrontEndConfiguration
					.getInstance().getPredictionLength());
			double utilization = Utilization.getClusterAverageUsed() * 2;
			double predictedUtilization = utilization + (utilization * change);
			double predictedQos = predictor.getAverageQos()
					+ predictor.getAverageQos() * change;

			System.out.println(predictedQos + "::::"
					+ predictor.getAverageQos());

			if (predictedQos > upperBound) {
				double scaleFactor = (predictedQos / targetQos) / 4.0;
				double targetFrequency = controller.getTotalFreq()
						+ scaleFactor * controller.getTotalFreq();
				try {
					scale = controller
							.scaleUpToFrequency(targetFrequency, true);
					if (scale != Scale.NONE) {
						this.proactiveElected++;
						System.out.println("Proactive Elected: "
								+ proactiveElected);
					}
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			} else if (predictedQos < lowerBound) {

				int count = (int) ((FrontEndConfiguration.getInstance()
						.getReactiveScaleDownTime() * 1000) / FrontEndConfiguration
						.getInstance().getReactorSleepTime());
				if (scaleDownCounter >= count) {
					double targetFrequency = controller.getTotalFreq();
					double scaleFactor = 1;
					if (predictedQos < extemeLowerBound) {
						// Assume we're just overprovisioned
						if (change > 0) {
							scaleFactor = (predictor.getAverageQos() / targetQos) / 6.0;
							// assume we're going down
						} else {
							scaleFactor = (predictedQos / targetQos) / 4.0;
						}
					} else {
						if (change > 0) {
							scaleFactor = (predictor.getAverageQos() / targetQos) / 12.0;
						} else {
							scaleFactor = (predictedQos / targetQos) / 10.0;
						}
					}
					targetFrequency = controller.getTotalFreq() - scaleFactor
							* controller.getTotalFreq();
					try {
						if (controller.oneCanScaleDown(false)) {
							scale = controller.scaleDownToFrequency(
									targetFrequency, false);
							if (scale == Scale.NONE) {
								scale = controller.scaleDownOne(false);
							}
						} else if (controller.oneCanScaleDown(true)) {
							scale = controller.scaleDownToFrequency(
									targetFrequency, true);
							if (scale == Scale.NONE) {
								scale = controller.scaleDownOne(true);
							}
						}
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					scaleDownCounter = 0;

				}
			}

			if (scale == Scale.ON) {
				this.timeToChangeScale = currentTime
						+ FrontEndConfiguration.getInstance().getTimeToOn();
				scaleDownCounter = 0;
			} else if (scale == Scale.OFF) {
				this.timeToChangeScale = currentTime
						+ FrontEndConfiguration.getInstance().getTimeToOff();
				scaleDownCounter = 0;
			} else if (scale == Scale.UP || scale == Scale.DOWN) {
				this.timeToChangeScale = currentTime
						+ FrontEndConfiguration.getInstance().getTimeToScale();
				scaleDownCounter = 0;
			}
			if (scale != Scale.NONE) {
				System.out.println("predictive");
			}
		}
	}

	@Override
	public void alertResponseListener(ReactorThread rt) {
		ServerController farm = ServerController.getInstance();
		FrontEndConfiguration fec = FrontEndConfiguration.getInstance();
		long currentTime = System.currentTimeMillis();
		double targetFrequency = farm.getTotalFreq() + farm.getTotalFreq()
				* rt.getPercentMissed();
		// +500 to give the predictor a chance to run
		if (this.timeToChangeScale + 500 < currentTime) {
			if (rt.getPercentMissed() > fec.getLateThresshold()) {
				this.reactiveElected++;
				System.out.println("Reactive Elected: " + this.reactiveElected);
				Scale scale;
				try {
					System.out.println("scalingup");
					scale = farm.scaleUpToFrequency(targetFrequency, true);
					if (scale == Scale.ON) {
						this.timeToChangeScale = currentTime
								+ FrontEndConfiguration.getInstance()
										.getTimeToOn();
					} else {
						this.timeToChangeScale = currentTime
								+ FrontEndConfiguration.getInstance()
										.getTimeToScale();
					}
					System.out.println("reactive");
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			if (rt.getMissed() == 0) {
				scaleDownCounter++;
			} else {
				scaleDownCounter = 0;
			}
		}
	}
}
